package pool

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

const (
	DefaultCapacity = 50               // 容量，最大的协程数量
	DefaultTimeOut  = 30 * time.Second // 任务执行的时间，超过该时间协程会退出执行该程序，返回空闲状态
	DefaultIdleTime = 10 * time.Minute // 协程空闲的时间，超过该时间会自动
)

type GoroutinePool struct {
	// 容量
	capacity int32
	// 当前的goroutine数量，包括正在运行的和空闲的
	currentRunCount int32
	// 可以工作的worker，这里存放的是空闲的worker，忙碌的worker会从这里去掉，忙完把自己注册回这里等待新的任务
	idleWorkers []*Worker
	// 通知是否关闭自己
	closed bool
	lock   sync.Mutex
	// 超时时间
	timeout time.Duration
	// 最大空闲时间
	idleTime time.Duration
}

func NewGoroutinePool(opts ...Option) *GoroutinePool {
	pool := &GoroutinePool{
		capacity:        DefaultCapacity,
		currentRunCount: 0,
		timeout:         DefaultTimeOut,
		closed:          false,
		idleTime:        DefaultIdleTime,
	}
	if len(opts) > 0 {
		for _, opt := range opts {
			opt(pool)
		}
	}
	pool.idleWorkers = make([]*Worker, 0, pool.capacity)
	return pool
}

// Submit 提交执行的任务
func (p *GoroutinePool) Submit(task f) error {
	if p.closed {
		return errors.New("pool already closed")
	}
	if task == nil {
		return errors.New("task is nil")
	}
	// 获取可以执行的worker
	w := p.getWorker()
	w.sendTask(task)
	return nil
}

// 获取一个可以执行的worker
func (p *GoroutinePool) getWorker() *Worker {
	var w *Worker
	// 标志，表示当前运行的worker数量是否已达容量上限
	waiting := false

	// 涉及从workers队列取可用worker，需要加锁
	p.lock.Lock()
	workers := p.idleWorkers
	n := len(workers) - 1
	// 当前worker队列为空(无空闲worker)
	if n < 0 {
		// 运行worker数目已达到该Pool的容量上限，置等待标志
		if p.currentRunCount >= p.capacity {
			waiting = true
			// 否则，运行数目加1
		} else {
			p.currentRunCount++
		}
		// 有空闲worker，从队列尾部取出一个使用
	} else {
		w = workers[n]
		workers[n] = nil
		p.idleWorkers = workers[:n]
	}
	p.lock.Unlock()

	// 阻塞等待直到有空闲worker
	if waiting {
		// 队列有空闲worker通知信号
		for {
			p.lock.Lock()
			workers = p.idleWorkers
			l := len(workers) - 1
			if l < 0 {
				p.lock.Unlock()
				time.Sleep(5 * time.Millisecond)
				continue
			}
			w = workers[l]
			workers[l] = nil
			p.idleWorkers = workers[:l]
			p.lock.Unlock()
			break
		}
		// 当前无空闲worker但是Pool还没有满，
		// 则可以直接新开一个worker执行任务
	} else if w == nil {
		w = &Worker{
			pool:     p,
			task:     make(chan f),
			timeout:  p.timeout,
			idleTime: p.idleTime,
		}
		w.run()
	}
	return w
}

func (p *GoroutinePool) putWorker(w *Worker) {
	if w == nil {
		return
	}
	p.lock.Lock()
	p.idleWorkers = append(p.idleWorkers, w)
	p.lock.Unlock()
}

func (p *GoroutinePool) removeWorker(w *Worker) {
	if w == nil {
		return
	}
	if len(p.idleWorkers) == 0 {
		return
	}

	index := -1
	p.lock.Lock()
	workers := p.idleWorkers
	for i, worker := range workers {
		if worker == w {
			index = i
			break
		}
	}
	if index == -1 {
		p.lock.Unlock()
		return
	}
	workers = append(workers[:index], workers[index+1:]...)
	p.idleWorkers = workers
	p.lock.Unlock()
}

func (p *GoroutinePool) Close() {
	p.closed = true
	// 如果还有等待的worker,放置空任务以关闭
	for {
		p.lock.Lock()
		workers := p.idleWorkers
		l := len(workers) - 1
		if l < 0 {
			p.lock.Unlock()
			break
		}
		p.lock.Unlock()
		w := p.getWorker()
		if w != nil {
			w.Close()
		}
	}
	// 清空切片
	p.idleWorkers = p.idleWorkers[:0]
}

// 信号的类型
type sig struct{}

// 执行的函数
type f func()

// Worker 对协程进行封装的worker,每一个worker绑定一个goroutine
type Worker struct {
	// 任务，一个任务的chan
	task chan f
	// 任务完成的channel
	pool *GoroutinePool
	// 超时时间
	timeout time.Duration
	// 最大空闲时间
	idleTime time.Duration
}

// 运行程序
func (w *Worker) run() {
	go func() {
		idleTime := time.After(w.idleTime)
		//等待任务执行
		for {
			select {
			case fData, ok := <-w.task:
				if !ok || w.pool.closed {
					// 关闭
					atomic.AddInt32(&w.pool.currentRunCount, -1)
					goto Loop
				}
				// 执行函数
				w.processWorker(fData)
				//再次检查是否关闭
				if w.pool.closed {
					atomic.AddInt32(&w.pool.currentRunCount, -1)
					goto Loop
				}
				// 将自己注册到缓冲区，方便发送任务
				w.pool.putWorker(w)
				// 重置空闲时间定时器
				idleTime = time.After(w.idleTime)
			case <-idleTime:
				w.pool.removeWorker(w)
				w.Close()
			}
		}
	Loop:
		//fmt.Println("worker exit")
	}()
}

func (w *Worker) processWorker(fData f) {
	if fData == nil {
		return
	}

	//借助于带超时的context来实现对函数的超时控制
	timeout := w.timeout
	if timeout < 0 {
		timeout = DefaultTimeOut
	}
	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel() //纯粹出于良好习惯，函数退出前调用cancel()
	workDone := make(chan sig, 1)
	go func() {
		defer func() {
			// 这里捕获异常，防止因为执行函数的错误，导致整个进程失败
			// 可以在这里增加日志
			err := recover()
			if err != nil {
				fmt.Printf("process error :%v \n", err)
			}
			workDone <- sig{}
		}()
		fData()
	}()
	select {
	case <-ctx.Done():
		fmt.Println("time out")
	case <-workDone:
	}
}

func (w *Worker) sendTask(task f) {
	w.task <- task
}

func (w *Worker) Close() {
	defer func() {
		if r := recover(); r != nil {
			fmt.Println(" closed channel err:", r)
		}
	}()
	close(w.task)
}

type Option func(pool *GoroutinePool)

func WithCapacity(capacity int32) Option {
	return func(pool *GoroutinePool) {
		if capacity <= 0 {
			capacity = DefaultCapacity
		}
		pool.capacity = capacity
	}
}

func WithTimeout(timeout time.Duration) Option {
	return func(pool *GoroutinePool) {
		if timeout <= 0 {
			timeout = DefaultTimeOut
		}
		pool.timeout = timeout
	}
}

func WithIdleTime(idleTime time.Duration) Option {
	return func(pool *GoroutinePool) {
		if idleTime <= 0 {
			idleTime = DefaultIdleTime
		}
		pool.idleTime = idleTime
	}
}
